package pipeline

import (
	"bufio"
	"net"
)

// 节点服务端数据写入到Network中
// 开启服务后，用goroutine等连接，避免创建pipeline阻塞
func NetworkSink(addr string, in <-chan int) {
	//net必须是面向流的网络："tcp"、"tcp4"、"tcp6"、"unix"或"unixpacket"
	listener, err := net.Listen("tcp", addr) //addr ip:port
	if err != nil {
		panic(err)
	}
	go func() { //不能等待阻塞
		for {
			conn, err := listener.Accept()
			if err != nil {
				continue
			}
			w := bufio.NewWriter(conn)
			WriteSink(w, in)
			w.Flush()    //使用bufio Writer最后一定要Flush把缓存数据发出去  defer
			conn.Close() //关闭
		}
		// defer listener.Close()
		// conn, err := listener.Accept()
		// if err != nil {
		// 	panic(err)
		// }
		// defer conn.Close()
		// w := bufio.NewWriter(conn)
		// WriteSink(w, in)
		// defer w.Flush()
	}()
}

// Network向节点服务端读取数据源
func NetworkSource(addr string) <-chan int {
	out := make(chan int)
	go func() {
		conn, err := net.Dial("tcp", addr)
		if err != nil {
			panic(err)
		}
		defer conn.Close()

		r := ReadSource(bufio.NewReader(conn), -1)
		for v := range r {
			out <- v
		}
		close(out)
	}()
	return out
}
